#include "AmqpWorker.h"
#include "StreamPuller.h"
#include "utils.hpp"
#include <unistd.h>
#include <signal.h>
#include "json.hpp"
#include "spdlog/spdlog.h"
#include "spdlog/cfg/env.h"
#include <exception>
#include <thread>
#include <iostream>
#include <fstream>

class Application
{
public:
	struct Properties {
		int maxThreads;
		int taskType;
		bool noWebServer;
		struct {
			std::string uri;
		} source;
		struct {
			std::string uri;
			std::string root;
		} target;
		struct {
			double fps, threshold, scale;
		} video_cap;
	};
	static Properties ParseArgs(int argc, char *argv[]);
	
	Application(int taskType, bool noWebServer=false, int maxThreads=0)
		: amqp_worker(taskType, noWebServer, maxThreads?maxThreads:10) {}
	
	void run(const Properties& args) {
		SPDLOG_INFO("Start running ...");
		_args = args;
		if (!onInit()) {
			SPDLOG_CRITICAL("Init failed !!!");
			return;
		}
		
		signal(SIGTERM, onSystemExit);
		signal(SIGQUIT, onSystemExit);
		signal(SIGINT, onSystemExit);
		
		while (!__system_terminate) {
			if (onLoop() != 0) {
				break;
			}
		}
		onExit();
		SPDLOG_INFO("End running");
	}

private:
	AmqpWorker amqp_worker;
	Properties _args;
	std::shared_ptr<SafetyQueue<TaskReply>> qt_reply = std::make_shared<SafetyQueue<TaskReply>>();
	
	static int __system_terminate;
	static void onSystemExit(int signum) {
		__system_terminate = signum;
	}
	
	bool onInit() {
		auto uri = parse_uri(_args.source.uri);
		if (uri.port == -1) uri.port = 5672;
		if (uri.auth.username.empty()) uri.auth.username = "guest";
		if (uri.auth.password.empty()) uri.auth.password = "guest";
		if (!amqp_worker.Connect(uri.host, uri.port, uri.auth.username, uri.auth.password)) {
			return false;
		}
		amqp_worker.Register();
		return true;
	}
	
	int onLoop() {
		amqp_worker.drain_events(20);
		
		std::string task_id, url;
		if (amqp_worker.pull_task(task_id, url)) {
			SPDLOG_INFO("pull task: {{taskId='{}', url='{}'}}", task_id.c_str(), url.c_str());
			process(task_id, url);
		}
		
		while (!qt_reply->empty()) {
			TaskReply reply;
			if (qt_reply->get(reply)) {
				SPDLOG_INFO("get reply: {{taskId='{}', code={}, message='{}'}}\n",
					reply.taskId.c_str(), reply.replyCode, reply.message.c_str());
				amqp_worker.finish_task(reply.taskId, reply.replyCode>=0);
			}
		}
		
		return 0;
	}
	
	void onExit() {
		amqp_worker.Unregister();
		amqp_worker.Close();
	}
	
	void process(const std::string& task_id, const std::string& url) {
		TaskRequest task{};
		task.taskId = task_id;
		task.source.url = url;
		task.source._fps = _args.video_cap.fps;
		task.source._threshold = _args.video_cap.threshold;
		task.source._scale = _args.video_cap.scale;
		task.target.uri = _args.target.uri;
		task.target._root = _args.target.root;
		auto q = qt_reply;
		
		std::thread([task, q]() {
			auto reply = StreamPuller(task);
			q->put(reply);
		}).detach();
	}
};

int Application::__system_terminate = 0;

Application::Properties Application::ParseArgs(int argc, char *argv[])
{
	using json = nlohmann::json;
	Properties args{};
	
	int opt, maxThreads = 0, taskType = 0;
	bool noWebServer = false;
	while ((opt = getopt(argc, argv, "p:w:t:")) != -1) {
		switch (opt) {
		case 'p': maxThreads = atoi(optarg); break;
		case 'w': noWebServer = !atoi(optarg); break;
		case 't': taskType = atoi(optarg); break;
		}
	}
	args.maxThreads = maxThreads;
	args.noWebServer = noWebServer;
	
	if (optind >= argc) {
		throw std::string("Not specify a json config file");
	}
	json j_cfg;
	std::ifstream(argv[optind]) >> j_cfg;
	
	args.taskType = taskType;
	args.source.uri = j_cfg["source"]["uri"];
	args.target.uri = j_cfg["target"]["uri"];
	args.target.root = j_cfg["target"]["root"];
	args.video_cap.fps = j_cfg["videoCapture"]["fps"];
	args.video_cap.threshold = j_cfg["videoCapture"]["threshold"];
	args.video_cap.scale = j_cfg["videoCapture"]["scale"];
	return args;
}

int main(int argc, char *argv[])
{
	spdlog::cfg::load_env_levels();
	
	Application::Properties args;
	std::string parse_error;
	try {
		args = Application::ParseArgs(argc, argv);
	} catch (const std::string& errmsg) {
		parse_error = errmsg;
	} catch (std::exception& e) {
		parse_error = e.what();
	}
	if (!parse_error.empty()) {
		SPDLOG_CRITICAL("parse args failed: {}", parse_error);
		std::cout << "Usage: " << argv[0]
		          << " [-p max_threads_count] [-w withWebServer] -t taskType json_config_file" << std::endl;
		return -1;
	}
	
	Application(args.taskType, args.noWebServer, args.maxThreads).run(args);
	return 0;
}
